package com.ffcs.crmd.tsp.server.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ffcs.crmd.tsp.api.dto.client.TspJobDetailThreadInfo;
import com.ffcs.crmd.tsp.api.dto.job.TspJobDetailShell;
import com.ffcs.crmd.tsp.api.dto.netty.RemotingCommand;
import com.ffcs.crmd.tsp.api.dto.task.TaskThreadPoolInfo;
import com.ffcs.crmd.tsp.api.dto.task.TspJobInstance;
import com.ffcs.crmd.tsp.api.netty.INettyConfigServer;
import com.ffcs.crmd.tsp.api.netty.INettyRemotingServer;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.common.enu.JobDetailSendType;
import com.ffcs.crmd.tsp.common.exception.TspException;
import com.ffcs.crmd.tsp.common.exception.TspRemotingException;
import com.ffcs.crmd.tsp.common.netty.RequestCode;
import com.ffcs.crmd.tsp.common.netty.ResponseCode;
import com.ffcs.crmd.tsp.core.util.XmlUtil;
import com.ffcs.crmd.tsp.server.api.Handler;
import com.ffcs.crmd.tsp.server.application.ApplicationManagerHolder;
import com.ffcs.crmd.tsp.server.jdbc.entity.TaskIdentity;
import com.ffcs.crmd.tsp.server.job.info.JobDetailShell;
import com.ffcs.crmd.tsp.server.job.service.JobRecivedCallBackService;
import com.ffcs.crmd.tsp.server.job.service.JobSendFailCallBackService;
import com.ffcs.crmd.tsp.server.manager.TaskRemotingManager;
import com.ffcs.crmd.tsp.server.netty.NettyRemotingServer;
import com.ffcs.crmd.tsp.server.netty.client.ClientKeepActiveListener;
import com.ffcs.crmd.tsp.server.netty.processor.CompleteJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateJobDetailParamProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateJobIdentityInfoProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateSliceGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateSliceProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.CreateTaskIdentityProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobDetailDeathProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobDetailParamProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobDetailRunningProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobIdentityInfoProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteSliceGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteSliceProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.DeleteTaskIdentityProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.HeartDataProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.InterruptJobDetaiRunninglProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.PauseJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.PauseJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.PauseJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryClusterInstanceProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryClusterProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailDeathProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailParamProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailRetryProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailRunningProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailThreadInfoProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailTrackProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobDetailWaitingProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobIdentityInfoProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QuerySliceGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QuerySliceProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryTaskConnectInfoProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.QueryTaskIdentityProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.ResetJobDetailDeathProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.ResetJobDetailRunningProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.ResumeJobDetailProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.ResumeJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.ResumeJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateJobDetailParamProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateJobGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateJobProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateSliceGroupProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateSliceProcessor;
import com.ffcs.crmd.tsp.server.netty.processor.UpdateTaskIdentityProcessor;
import com.ffcs.crmd.tsp.server.netty.task.TaskKeepActiveListener;
import com.ffcs.crmd.tsp.server.netty.task.TaskRegisterInfo;
import com.ffcs.crmd.tsp.server.util.PublicMethodHelper;

import io.netty.channel.Channel;

/**
 * 
 * 功能说明:通信处理中心
 *
 * @author ZHONGFUHUA
 * 
 * @since 1.0.0-SNAPSHOT
 *
 */
public class RemotingHandler implements Handler {
    
    /**
     * 日志
     */
    private final static Logger      LOG                 = LoggerFactory.getLogger(LoggerName.ServerLoggerName);
    
    /**
     * 通信服务端
     */
    private INettyRemotingServer     remotingServer;
    
    /**
     * 任务端保持活跃监听服务
     */
    private TaskKeepActiveListener   taskKeepActiveListener;
    
    /**
     * 客户端保持活跃监听服务
     */
    private ClientKeepActiveListener clientKeepActiveListener;
    
    /**
     * 数据源处理中心
     */
    private DataBaseHandler          dataBaseHandler;
    
    /**
     * 执行端通信管理者
     */
    private TaskRemotingManager      taskRemotingManager = ApplicationManagerHolder.getBean("taskRemotingManager", TaskRemotingManager.class);
    
    
    public RemotingHandler(INettyConfigServer serverConfig) {
        this.taskKeepActiveListener = new TaskKeepActiveListener();
        this.clientKeepActiveListener = new ClientKeepActiveListener();
        this.remotingServer = new NettyRemotingServer(serverConfig, taskKeepActiveListener, clientKeepActiveListener);
    }
    
    @Override
    public void start() throws TspException {
        this.remotingServer.registerProcessor(RequestCode.COMPLETE_JOB_DETAIL, new CompleteJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_JOB_DETAIL, new CreateJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_JOB_GROUP, new CreateJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_JOB, new CreateJobProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_SLICE_GROUP, new CreateSliceGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_SLICE, new CreateSliceProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_JOB_IDENTITY_INFO, new CreateJobIdentityInfoProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_TASK_IDENTITY, new CreateTaskIdentityProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.CREATE_JOB_DETAIL_PARAM, new CreateJobDetailParamProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.DELETE_JOB_DETAIL, new DeleteJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB_GROUP, new DeleteJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB, new DeleteJobProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_SLICE_GROUP, new DeleteSliceGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_SLICE, new DeleteSliceProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB_IDENTITY_INFO, new DeleteJobIdentityInfoProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB_DETAIL_DEATH, new DeleteJobDetailDeathProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_TASK_IDENTITY, new DeleteTaskIdentityProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB_DETAIL_RUNNING, new DeleteJobDetailRunningProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.DELETE_JOB_DETAIL_PARAM, new DeleteJobDetailParamProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.RESET_JOB_DETAIL_RUNNING, new ResetJobDetailRunningProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.RESET_JOB_DETAIL_DEATH, new ResetJobDetailDeathProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.HEART_BEAT, new HeartDataProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.INTERRUPT_JOB_DETAIL_RUNNING, new InterruptJobDetaiRunninglProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.PAUSE_JOB_DETAIL, new PauseJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.PAUSE_JOB_GROUP, new PauseJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.PAUSE_JOB, new PauseJobProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.QUERY_CLUSTER_INSTANCE, new QueryClusterInstanceProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_CLUSTER, new QueryClusterProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_THREAD_INFO, new QueryJobDetailThreadInfoProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_DEATH, new QueryJobDetailDeathProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_TRACK, new QueryJobDetailTrackProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_RUNNING, new QueryJobDetailRunningProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_WAITING, new QueryJobDetailWaitingProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_PARAM, new QueryJobDetailParamProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL, new QueryJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_GROUP, new QueryJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB, new QueryJobProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_SLICE_GROUP, new QuerySliceGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_SLICE, new QuerySliceProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_TASK_CONNECT_INFO, new QueryTaskConnectInfoProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_IDENTITY_INFO, new QueryJobIdentityInfoProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_JOB_DETAIL_RETRY, new QueryJobDetailRetryProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.QUERY_TASK_IDENTITY, new QueryTaskIdentityProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.RESUME_JOB_DETAIL, new ResumeJobDetailProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.RESUME_JOB_GROUP, new ResumeJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.RESUME_JOB, new ResumeJobProcessor(this, dataBaseHandler))
            
            .registerProcessor(RequestCode.UPDATE_JOB_DETAIL_PARAM, new UpdateJobDetailParamProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.UPDATE_JOB_GROUP, new UpdateJobGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.UPDATE_JOB, new UpdateJobProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.UPDATE_SLICE_GROUP, new UpdateSliceGroupProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.UPDATE_SLICE, new UpdateSliceProcessor(this, dataBaseHandler))
            .registerProcessor(RequestCode.UPDATE_TASK_IDENTITY, new UpdateTaskIdentityProcessor(this, dataBaseHandler));
        
        this.taskKeepActiveListener.start();
        this.clientKeepActiveListener.start();
        try {
            this.remotingServer.start();
            LOG.info("SERVER-REMOTING: Remoting handler start ok.");
        } catch (TspRemotingException e) {
            throw new TspException(e);
        }
    }
    
    @Override
    public void shutdown() throws TspException {
        this.remotingServer.shutdown();
        this.taskKeepActiveListener.shutdown();
        this.clientKeepActiveListener.shutdown();
        LOG.info("SERVER-REMOTING: Remoting handler shutdown ok.");
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     * @param jobDetailShell
     * @param sendJobDetailType
     * @return
     */
    public boolean send(Channel channel, JobDetailShell jobDetailShell, JobDetailSendType sendJobDetailType) {
        TspJobInstance tspJobInstance = jobDetailShell.getContext().getTspJobInstance();
        RemotingCommand remotingCommand = null;
        TaskRegisterInfo taskRegisterInfo = taskRemotingManager.getTaskRegisterInfo(channel);
        if (PublicMethodHelper.validateOldVersion(taskRegisterInfo)) {
            switch (tspJobInstance.getJobType()) {
                case NORMAL:
                    remotingCommand = PublicMethodHelper.createShellCommandOld(jobDetailShell, RequestCode.RUN_JOB_DETAIL);
                    break;
                case STATEFUL:
                    remotingCommand = PublicMethodHelper.createShellCommandOld(jobDetailShell, RequestCode.RUN_JOB_DETAIL_STATFUL);
                    break;
            }
        } else {
            switch (tspJobInstance.getJobType()) {
                case NORMAL:
                    remotingCommand = PublicMethodHelper.createShellCommand(jobDetailShell, RequestCode.RUN_JOB_DETAIL);
                    break;
                case STATEFUL:
                    remotingCommand = PublicMethodHelper.createShellCommand(jobDetailShell, RequestCode.RUN_JOB_DETAIL_STATFUL);
                    break;
            }
        }
        
        int opaque = jobDetailShell.getRemotingOpaque();
        jobDetailShell.setRemotingOpaque(remotingCommand.getOpaque());
        jobDetailShell.updateSendTime();
        
        JobRecivedCallBackService callBack = new JobRecivedCallBackService(jobDetailShell, channel, taskRegisterInfo);
        JobSendFailCallBackService sendFailcallBack = new JobSendFailCallBackService(jobDetailShell, channel, taskRegisterInfo, sendJobDetailType);
        remotingServer.invokeAsync(channel, remotingCommand, callBack, sendFailcallBack);
        if (opaque != 0) {
            remotingServer.removeResponse(opaque);
        }
        return true;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     * @param channel
     * @return
     */
    public boolean shutdown(JobDetailShell jobDetailShell, Channel channel) {
        TspJobInstance tspJobInstance = jobDetailShell.getContext().getTspJobInstance();
        try {
            RemotingCommand remotingCommand = PublicMethodHelper.createShellCommand(jobDetailShell, RequestCode.INTERRUPT_JOB_DETAIL_RUNNING);
            RemotingCommand result = remotingServer.invokeSync(channel, remotingCommand);
            if (result.getCode() == ResponseCode.SUCCESS) {
                return Boolean.valueOf(new String(result.getBody()));
            }
        } catch (TspRemotingException e) {
            LOG.error("SERVER-SCHEDULER: shutdown a {} job detail[{}] id=[{}] in processing [{}] exception:", tspJobInstance.getJobType(),
                tspJobInstance.getJobKey(), jobDetailShell.getId(), channel, e);
        }
        return false;
    }
    
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     * @param channel
     * @return
     */
    public boolean shutdownFinish(JobDetailShell jobDetailShell, Channel channel) {
        TspJobInstance tspJobInstance = jobDetailShell.getContext().getTspJobInstance();
        try {
            RemotingCommand remotingCommand = PublicMethodHelper.createShellCommand(jobDetailShell, RequestCode.INTERRUPT_JOB_DETAIL_RUNNING_FINISH);
            RemotingCommand result = remotingServer.invokeSync(channel, remotingCommand);
            if (result.getCode() == ResponseCode.SUCCESS) {
                return Boolean.valueOf(new String(result.getBody()));
            }
        } catch (TspRemotingException e) {
            LOG.error("SERVER-SCHEDULER: shutdown a {} job detail finish[{}] id=[{}] in processing [{}] exception:", tspJobInstance.getJobType(),
                tspJobInstance.getJobKey(), jobDetailShell.getId(), channel, e);
        }
        return false;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param taskRegisterInfo
     * @param taskIdentity
     * @param channel
     */
    public void updateThreadPool(TaskRegisterInfo taskRegisterInfo, TaskIdentity taskIdentity, Channel channel) {
        if (taskIdentity != null) {
            RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.UPDATE_THREAD_POOL);
            if (!PublicMethodHelper.validateOldVersion(taskRegisterInfo)) {
                TaskThreadPoolInfo taskThreadPoolInfo = new TaskThreadPoolInfo(taskIdentity.getBatchThreadNum(), taskIdentity.getThreadNum());
                taskThreadPoolInfo.setJobProcessThreadPriorityWeight(PublicMethodHelper.getThreadPriorityWeight(taskIdentity));
                taskThreadPoolInfo.setJobProcessBusinessThreadPriorityWeight(PublicMethodHelper.getBacthThreadPriorityWeight(taskIdentity));
                remotingCommand.setBody(XmlUtil.toXml(taskThreadPoolInfo).getBytes());
            } else {
                remotingCommand.setBody(String.valueOf(taskIdentity.getBatchThreadNum()).getBytes());
            }
            remotingServer.invokeOneway(channel, remotingCommand);
        }
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param jobDetailShell
     * @param channel
     * @return
     */
    public TspJobDetailThreadInfo queryThreadInfo(JobDetailShell jobDetailShell, Channel channel) {
        TspJobDetailShell tspJobDetailShell = new TspJobDetailShell();
        tspJobDetailShell.setPriority(jobDetailShell.getContext().getTspJobInstance().getPriority());
        tspJobDetailShell.setId(jobDetailShell.getId());
        tspJobDetailShell.setContext(jobDetailShell.getContext());
        try {
            RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.QUERY_JOB_DETAIL_THREAD_INFO);
            requestCommand.setBody(XmlUtil.toXml(tspJobDetailShell).getBytes());
            RemotingCommand result = remotingServer.invokeSync(channel, requestCommand);
            if (result.getCode() == ResponseCode.SUCCESS) {
                TspJobDetailThreadInfo tspJobDetailThreadInfo = XmlUtil.fromXml(new String(result.getBody()), TspJobDetailThreadInfo.class);
                return tspJobDetailThreadInfo;
            }
        } catch (TspRemotingException e) {
            LOG.error("SERVER-SCHEDULER: query a job detail[{}]  id=[{}] thread info in processing [{}] exception:", jobDetailShell.getJobKey(),
                jobDetailShell.getId(), channel, e);
        }
        return null;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param channel
     */
    public boolean testConnect(Channel channel){
        if (PublicMethodHelper.validateChannel(channel)) {
            RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.TEST_CONNECT);
            requestCommand.setBody("test connect".getBytes());
            remotingServer.invokeOneway(channel, requestCommand);
            return true;
        }
        return false;
    }
    
    /**
     * 
     * 功能说明:
     * 
     * @param opaque
     * @return
     */
    public void releaseResponse(int opaque) {
        remotingServer.removeResponse(opaque);
    }
    
    
    /**
     * 功能说明: 设置数据源处理中心
     *
     * @param dataBaseHandler 数据源处理中心 
     */
    public void setDataBaseHandler(DataBaseHandler dataBaseHandler) {
        this.dataBaseHandler = dataBaseHandler;
    }
    
}
